/**
 * 
 */
package edu.umd.clip.jobs;

import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.*;
import java.util.concurrent.*;
/**
 * @author Denis Filimonov <den@cs.umd.edu>
 *
 */
public class JobManager implements Runnable {
	private static JobManager instance;
	private final Worker[] workers;
	private final ThreadGroup workersThreadGroup;
	private final ReentrantLock workersLock;
	private final Queue<Job> jobQueue;
	private final ConcurrentLinkedQueue<Worker> freeWorkers;
	private final ConcurrentLinkedQueue<Worker> tempWorkers;
	private final Condition hasJobs;
	private final Condition hasWorkers;
	private final int nrWorkers;
	private final AtomicInteger workersToFire;
	
	private JobManager(int nrWorkers, int jobQueueSize) {
		this.nrWorkers = nrWorkers;
		workersLock = new ReentrantLock();
		hasJobs = workersLock.newCondition();
		hasWorkers = workersLock.newCondition();
		
		workersThreadGroup = new ThreadGroup("workers group");
		// TODO: ArrayBlockingQueue need a workaround in addJob and addBatch
		jobQueueSize = 0;
		jobQueue = jobQueueSize > 0 ? new ArrayBlockingQueue<Job>(jobQueueSize) : new ConcurrentLinkedQueue<Job>();
		workers = new Worker[nrWorkers];
		freeWorkers = new ConcurrentLinkedQueue<Worker>();
		for(int i=0; i<nrWorkers; ++i) {
			Worker worker = new Worker(this, "Worker #"+Integer.toString(i)); 
			worker.setDaemon(true);
			workers[i] = worker;
			freeWorkers.add(worker);
		}
		
		tempWorkers = new ConcurrentLinkedQueue<Worker>();
		workersToFire = new AtomicInteger();
	}
	
	static class JobComparator implements Comparator<Job> {
		public int compare(Job job1, Job job2) {
			return (int)Math.signum(job1.getPriority() - job2.getPriority());
		}
	}
	
	public static void initialize(int nrWorkers) {
		initialize(nrWorkers, 0);
	}
	
	public static void initialize(int nrWorkers, int jobQueueSize) {
		if (instance == null) {
			instance = new JobManager(nrWorkers, jobQueueSize);
		}
	}
	
	protected void finished(Worker worker) {
		Job job = worker.getCurrentJob();
		//System.out.println("Job "+job.getName()+" has finished.");
		job.getGroup().jobFinished(job);
		
		boolean wasEmpty = freeWorkers.isEmpty();
		
		freeWorkers.add(worker);
		if (wasEmpty) {
			try {
				workersLock.lock();
				hasWorkers.signal();
			} finally {
				workersLock.unlock();
			}
		}
	}
	
	public void addBatch(JobGroup group, Collection<Job> jobs) {
		for(Job job : jobs) {
			group.addJob(job);
		}
		if (Thread.currentThread() instanceof Worker) {
			for(Job job : jobs) {
				if (!jobQueue.offer(job)) {
					// will block the current worker, need to put a temporary worker instead
					goingToBlock();
					jobQueue.offer(job);
					goingToUnBlock();
				}
			}
		} else {
			for(Job job : jobs) {
				jobQueue.offer(job);
			}
		}
		try {
			workersLock.lock();
			hasJobs.signal();
		} finally {
			workersLock.unlock();
		}
	}
	
	public void addJob(JobGroup group, Job job) {
		group.addJob(job);
		
		boolean wasEmpty = jobQueue.isEmpty();
		
		if (Thread.currentThread() instanceof Worker) {
			if (!jobQueue.offer(job)) {
				// will block the current worker, need to put a temporary worker instead
				goingToBlock();
				jobQueue.offer(job);
				goingToUnBlock();
			}
		} else {
			jobQueue.offer(job);
		}
		if (wasEmpty) {
			try {
				workersLock.lock();
				hasJobs.signal();
			} finally {
				workersLock.unlock();
			}
		}
	}
	
	public void addJob(Job job) {
		if (job.getGroup() != null) {
			addJob(job.getGroup(), job);
			return;
		}
		if (Thread.currentThread() instanceof Worker) {
			// if the job was submitted from another job add it to the same JobGroup
			Job parentJob = ((Worker) Thread.currentThread()).getCurrentJob();
			addJob(parentJob.getGroup(), job);
		} else {
			addJob(createJobGroup("anonymous-"+job.getName()), job);
		}
	}
	
	/* (non-Javadoc)
	 * @see java.lang.Runnable#run()
	 */
	public void run() {
		final long maxDelay = TimeUnit.MILLISECONDS.toNanos(100);
		
		for(Worker worker : workers) {
			worker.start();
		}
		while(!Thread.currentThread().isInterrupted()) {
			workersLock.lock();
			try {
				while(jobQueue.isEmpty()) {
					// jobs may be added after isEmpty() check, so use polling
					hasJobs.awaitNanos(maxDelay);
				}
				
				while(freeWorkers.isEmpty()) {
					hasWorkers.awaitNanos(maxDelay);
				}

				fireWorkers();
				
			} catch(InterruptedException e) {
				e.printStackTrace();
				break;
			} finally {
				workersLock.unlock();
			}

			while(!(freeWorkers.isEmpty() || jobQueue.isEmpty())) {
				Job job = jobQueue.remove();
				Worker worker = freeWorkers.remove();
				//System.out.println("Scheduling job "+job.toString());
				worker.postJob(job);
			}
		}
	}

	private void hireTempWorker() {
		Worker worker = tempWorkers.poll();
		if (worker == null) {
			worker = new Worker(this, "temp worker");
			worker.setDaemon(true);
			worker.start();
		}
		
		boolean wasEmpty = freeWorkers.isEmpty();
		freeWorkers.add(worker);

		if (wasEmpty) {
			try {
				workersLock.lock();
				hasWorkers.signal();
			} finally {
				workersLock.unlock();
			}
		}
	}
	
	private void fireWorkers() {
		// workersLock must be held
		while(workersToFire.intValue() > 0 && !freeWorkers.isEmpty()) {
			int newNumWorkers = workersToFire.decrementAndGet();
			if (newNumWorkers < 0) {
				// put back
				workersToFire.incrementAndGet();
				break;
			}
			Worker worker = freeWorkers.remove();
			tempWorkers.add(worker);
		}
	}
	
	protected void goingToBlock() {
		if (Thread.currentThread() instanceof Worker) {
			// will block the current worker, need to put a temporary worker instead
			hireTempWorker();
		}
	}
	
	protected void goingToUnBlock() {
		if (Thread.currentThread() instanceof Worker) {
			workersToFire.incrementAndGet();
		}
	}
	
	public JobGroup createJobGroup(String name) {
		return new JobGroup(name);
	}
	/**
	 * @return the workersThreadGroup
	 */
	protected ThreadGroup getWorkersThreadGroup() {
		return workersThreadGroup;
	}

	public int getNumWorkers() {
		return nrWorkers;
	}
	
	/**
	 * @return the instance
	 */
	public static JobManager getInstance() {
		return instance;
	}
}
